package reader

import (
	"context"
	"fmt"

	"v8s-ssp-proxy/internal/configuration"
	"v8s-ssp-proxy/internal/types"
	"v8s/logger"

	"github.com/Shopify/sarama"

	"go.uber.org/zap"
)

type Reader struct {
	log *zap.SugaredLogger

	consumer sarama.ConsumerGroup
	topic    string
}

func NewReader(properties configuration.KafkaProperties) *Reader {
	config := sarama.NewConfig()

	tlsConfig, err := properties.TLS.ToConfig()
	if err != nil {
		panic(fmt.Errorf("failed to read tls config: %w", err))
	}

	if tlsConfig != nil {
		config.Net.TLS.Enable = true
		config.Net.TLS.Config = tlsConfig
	}

	config.Version = sarama.V2_0_0_0
	config.ClientID = "sup"

	consumer, err := sarama.NewConsumerGroup(properties.Addresses, properties.GroupID, config)
	if err != nil {
		panic(fmt.Errorf("failed to init connsumer group: %w", err))
	}

	return &Reader{
		log: logger.Logger().Named("mq_reader").Sugar(),

		consumer: consumer,
		topic:    properties.RequestTopic,
	}
}

func (r *Reader) Read(ctx context.Context, msgChan chan types.SSPPartnerMessage) {
	c := &consumer{r.log, msgChan, make(chan bool)}

	for {
		err := r.consumer.Consume(ctx, []string{r.topic}, c)
		if err != nil {
			r.log.Errorf("failed to consume: %s", err)
		}

		if ctx.Err() != nil {
			return
		}

		c.ready = make(chan bool)
	}
}
